package playground.fs.green;

/*-
 * #%L
 * A File Scanner App
 * %%
 * Copyright (C) 2017 - 2020 OSGL (Open Source General Library)
 * %%
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * #L%
 */

import org.osgl.$;
import org.osgl.Lang;
import org.osgl.util.*;
import playground.fs.LogFileScanner;

import java.io.File;
import java.util.*;
import java.util.concurrent.*;

public class GreenLogFileScannerOsglBigLines implements LogFileScanner {

    private static class LineCollector extends  $.Visitor<String> {

        private final ConcurrentMap<String, Integer> lineCounter = new ConcurrentHashMap<>();

        @Override
        public void visit(String s) throws Lang.Break {
            lineCounter.compute(s, (line, count) -> null == count ? 1 : ++count);
        }

        public List<String> output() {
            SortedMap<Integer, List<String>> sortedResult = new TreeMap<>();
            for (Map.Entry<String, Integer> entry : lineCounter.entrySet()) {
                sortedResult.compute(-1 * entry.getValue(), (k, v) -> {
                    if (null == v) v = new ArrayList<>();
                    v.add(entry.getKey());
                    return v;
                });
            }
            List<String> retList = new ArrayList<>();
            for (Map.Entry<Integer, List<String>> entry : sortedResult.entrySet()) {
                List<String> sorted = entry.getValue();
                sorted.sort($.F.reverseOrder());
                for (String line : sorted) {
                    retList.add(S.buffer().a(-1 * entry.getKey()).a(" ").a(line).toString());
                }
            }
            return retList;
        }
    }

    private static class FileScanner implements Runnable {

        private final BigLines bigLines;
        private final LineCollector lineCollector;
        private final CountDownLatch countDownLatch;
        private final String keyword;

        FileScanner(File logFile, String keyword, LineCollector lineCollector, CountDownLatch latch) {
            this.bigLines = new BigLines(logFile);
            this.lineCollector = $.requireNotNull(lineCollector);
            this.countDownLatch = $.requireNotNull(latch);
            this.keyword = S.requireNotBlank(keyword);
        }

        @Override
        public void run() {
            try {
                C.seq(bigLines).filter(s -> s.contains(keyword)).accept(lineCollector);
            } finally {
                countDownLatch.countDown();
            }
        }

    }

    @Override
    public List<String> doScan(File logFileDir, String keyword) throws Exception {
        long ms = $.ms();
        ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        File[] files = logFileDir.listFiles(f -> f.getName().endsWith(".log"));
        if (null == files) return C.list();
        LineCollector lineCollector = new LineCollector();
        CountDownLatch countDownLatch = new CountDownLatch(files.length);
        for (File file : files) executorService.submit(new FileScanner(file, keyword, lineCollector, countDownLatch));
        countDownLatch.await();
        executorService.shutdownNow();
        return lineCollector.output();
    }

}
